package top.pcat;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import top.pcat.codecs.MessageCodec;
import top.pcat.config.KafkaConfig;
import top.pcat.config.RedisConfig;
import top.pcat.config.ServerConfig;
import top.pcat.config.ZookeeperConfig;
import top.pcat.handle.NettyServerHandler;
import top.pcat.util.KafkaConsumerUtil;
import top.pcat.util.KafkaProducerUtil;
import top.pcat.util.RedisPoolUtil;
import top.pcat.util.ZookeeperUtil;

import java.io.IOException;

@Slf4j
public class Server {

    static EventLoopGroup bossGroup = new NioEventLoopGroup();
    static EventLoopGroup workerGroup = new NioEventLoopGroup();

    public static void main(String[] args) throws IOException {

        // 读取参数
        ServerConfig.serverHost = args[0];

        // 生成服务器id
        log.info("服务器id为");
        //ServerConfig.serverId =  NanoIdUtils.randomNanoId();
        ServerConfig.serverId = ServerConfig.serverHost;
        log.info("服务器id为"+ServerConfig.serverId);

        // zookeeper
        log.info("读取zookeeper配置文件");
        ZookeeperConfig zookeeperConfig = new ZookeeperConfig();
        new Thread(() -> {
            if(new ZookeeperUtil(zookeeperConfig, ServerConfig.serverHost).register())
                log.info("zookeeper注册成功");
            else
                log.info("zookeeper注册失败");
        }).start();

        // redis
        new RedisPoolUtil(new RedisConfig());

        // 清除上次服务器残留数据
        log.info("清除上次服务器残留数据");
        Jedis jedis = RedisPoolUtil.getJedis();
        jedis.del("server:"+ServerConfig.serverHost+":online_users");

        // Kafka Producer
        log.info("启动kafka Producer连接");
        new Thread(() -> {
            new KafkaProducerUtil(new KafkaConfig());
        }).start();

        // netty
        new Thread(() -> {
            log.info("启动netty连接");
            nettyInit();
        }).start();


        // kafka Consumer
        log.info("启动kafka Consumer连接");
        new Thread(() -> {
            log.info("启动kafka消息接收");
            KafkaConsumerUtil kafkaConsumerUtil = new KafkaConsumerUtil(new KafkaConfig());
            kafkaConsumerUtil.receiveMsgByKafka("test");
        }).start();


        //关闭监听
        log.info("启动关闭监听");
        Runtime.getRuntime().addShutdownHook(new Thread(() -> new ZookeeperUtil(zookeeperConfig, ServerConfig.serverHost).remove()));
    }

    public static void nettyInit()  {

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128) //链接缓冲池的大小
                    .childOption(ChannelOption.SO_KEEPALIVE, true) //维持链接的活跃，清除死链接
                    .childOption(ChannelOption.TCP_NODELAY, true) //关闭延迟发送
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new LengthFieldBasedFrameDecoder(1024000, 0, 4, 0, 0));
                            pipeline.addLast(new MessageCodec());
                            pipeline.addLast(new NettyServerHandler());
                        }
                    });

            log.info("来自服务器的消息 ----->  "+"服务端"+ ServerConfig.serverHost+"开始启动... ");

            ChannelFuture channelFuture = bootstrap.bind(Integer.parseInt(ServerConfig.serverHost.split(":")[1])).sync();
            channelFuture.addListener(future -> {
                if(future.isSuccess()) log.info("来自服务器的消息 ----->  服务器正在启动...");
                if(future.isDone()) log.info("来自服务器的消息 ----->  "+"服务器启动成功...OK");
            });

            channelFuture.channel().closeFuture().sync();
            channelFuture.addListener(future -> {
                if(future.isCancelled()) log.info("来自服务器的消息 ----->  "+"服务器正在关闭..");
                if(future.isCancellable()) log.info("来自服务器的消息 ----->  "+"服务器已经关闭..OK");
            });

        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            down();
        }
    }

    public static void down(){
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

